package io.sqrtqiezi.spark.streaming.kafka

import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object KafkaDStream1 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setAppName("spark kafka consumer")
      .setMaster("local[2]")
    val ssc = new StreamingContext(conf, Seconds(2))

    val kafkaParams: Map[String, Object] = getKafkaConsumerParameters("spark-consumer1")

    val topics: Array[String] = Array("topic-spark")

    val dstream = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
    )

    dstream.foreachRDD{(rdd, time) =>
      if (!rdd.isEmpty()) {
        println(s"****** rdd.count = ${rdd.count()}; time = $time ******")
      }
    }

    ssc.start()
    ssc.awaitTermination()
  }

  def getKafkaConsumerParameters(groupId: String): Map[String, Object] = {
    Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "izuf66rzw9m809spfn9yuaz:9092,izuf66rzw9m809spfn9yubz:9092,izuf66rzw9m809spfn9yucz:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> groupId,
      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest",
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean)
    )
  }
}
